package com.desire.公平调度队列;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author desire
 */
public class ConsumerTwo {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 配置连接信息  类似于 MySQL连接信息
        // 设置用户名 密码
        factory.setUsername("admin");
        factory.setPassword("123");
        // 设置mq服务器地址 默认是 localhost
        factory.setHost("116.63.151.50");
        // 设置高级消息队列amqp 端口号
        factory.setPort(5672);
        // 设置虚拟主机，默认是 /
        factory.setVirtualHost("erp");

        // 获取连接
        Connection connection = factory.newConnection();
        // 在连接上创建信道
        Channel channel = connection.createChannel();
        // 声明队列名称
        String queueName = "work_queue";
        channel.queueDeclare(queueName, true, false, false, null);
        // 同一时刻服务器只会发送一条消息给消费者
        channel.basicQos(1);
        System.out.println("消费者二等待处理消息：");
        // ---------------------------------------------
        // 从队列中读取消息   queue -队列的名称autoAck - true，如果服务器认为消息发送后被确认;如果服务器应该期望得到显式的确认，则为false
        channel.basicConsume(queueName, false, (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            try {
                System.out.println("消费者二开始处理消息：" + msg);
                doWork(msg);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 手动确认 告诉mq服务器消息处理完比 可以接收下一条消息
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
            System.out.println("完毕");
        }, (CancelCallback) null);
        //关闭连接
//        connection.close();
    }

    static void doWork(String task) {
        for (char c : task.toCharArray()) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
